package org.example.MapFunction;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.training.exercises.common.datatypes.EnrichedRide;
import org.apache.flink.training.exercises.common.datatypes.TaxiRide;
import org.apache.flink.util.Collector;
import org.example.RideCleansingExercise;

/**
 * @Author hujf
 * @Date 2022/8/24 14:09
 * @Description TODO
 */
public class NYCEnrichment implements FlatMapFunction<TaxiRide, EnrichedRide> {
    @Override
    public void flatMap(TaxiRide value, Collector<EnrichedRide> out) throws Exception {
        FilterFunction<TaxiRide> valid = new RideCleansingExercise.NYCFilter();
        if (valid.filter(value)) {
            out.collect(new EnrichedRide(value));
        }
    }
}
